package com.alit.face.koala.mq.apacheapollo;

import java.io.UnsupportedEncodingException;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;

import com.alit.face.koala.projectvariable.PVK;
import com.litong.utils.projectvariable.PVU;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class MQTTService {
  private MqttClient client;
  // 连接选项
  private MqttConnectOptions connectOptions;
  // private String server = "tcp://47.92.127.210";
  // private int port = 61613;

  // clientId和messageListener由外部类传入
  private String clientId = "";
  private MessageListener listener;

  // mqtt回调类
  private MqttCallbackExtended mqttCallbackExtended = new MqttCallbackExtended() {
    /**
     * 触发时机:连接陈成功后
     */
    @Override
    public void connectComplete(boolean arg0, String arg1) {
      listener.mqttConnection();
    }

    /**
     * 触发时机:收到boker返回的消息
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
      String string = new String(message.getPayload(), "utf-8");
      listener.handleMessage(topic, string);
    }

    /*
     * 触发时机:连接断开后
     */
    @Override
    public void connectionLost(Throwable arg0) {
      log.info("连接已经断开:" + arg0.getMessage());
      arg0.printStackTrace();
      listener.mqttClose();
    }

    /**
     * 接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
    }
  };

  public MQTTService(String clientId, MessageListener messageListener) {
    this.clientId = clientId;
    this.listener = messageListener;
    this.init();
  }

  /**
   * 初始化Client
   */
  private void init() {
    // 设置参数的值
    String server = PVU.getValue(PVK.apolloServer);
    int port = Integer.parseInt(PVU.getValue(PVK.apolloPort));
    String MQTCPURL = server + ":" + port;
    String username = PVU.getValue(PVK.apolloUsername);
    String passward = PVU.getValue(PVK.apolloPassword);

    try {
      client = new MqttClient(MQTCPURL, clientId);
    } catch (MqttException e) {
      e.printStackTrace();
    }

    // 设置回调对象
    client.setCallback(mqttCallbackExtended);
    // 实例化连接选项
    connectOptions = new MqttConnectOptions();
    // 清除缓存
    connectOptions.setCleanSession(true);
    // 设置超时时间：10秒
    connectOptions.setConnectionTimeout(10);
    // 心跳包发送间隔：20秒
    connectOptions.setKeepAliveInterval(20);
    connectOptions.setUserName(username);
    connectOptions.setPassword(passward.toCharArray());

    String topic = "test";
    String message = "{\"terminal_id\":\"" + clientId + "\"}";
    Integer qos = 2;
    Boolean retained = false;
    boolean doConnect = true;
    if ((!message.equals("")) || (!topic.equals(""))) {
      try {
        // 设置最后的遗嘱
        connectOptions.setWill(topic, message.getBytes(), qos.intValue(), retained.booleanValue());
      } catch (Exception e) {
        doConnect = false;
        e.printStackTrace();
      }
    }
    if (doConnect) {
      doClientConnection();
    }
  }

  /**
   * 连接
   */
  private void doClientConnection() {
    if (!client.isConnected()) {
      try {
        client.connect(connectOptions);
      } catch (MqttException e) {
        e.printStackTrace();
      }
    }
  }

  /**
   * 发送消息到主题
   */
  public boolean sendMessage(String topic, String msg, int qos) {
    boolean retaines = false;
    try {
      if (client != null && !client.isConnected()) {
        long start = System.currentTimeMillis();
        reconnect(3, start);
      }
      if (client != null && !client.isConnected()) {
        log.error("重新连接3次消息机,还是连接失败,检查消息机是否启动");
      }
      MqttTopic topicClient = client.getTopic(topic);
      if (topicClient != null) {
        // 向Broker中的指定主题发送消息
        topicClient.publish(msg.getBytes("utf-8"), qos, retaines);
        return true;
      }
    } catch (MqttException e) {
      log.error("发送消息出现异常:" + e.getMessage() + ",topic:" + topic);
    } catch (UnsupportedEncodingException e) {
      log.info("发送消息出现异常UnsupportedEncodingException" + e.getMessage());
    }
    return false;
  }

  private void reconnect(int count, long start) {
    for (int i = 0; i < count; i++) {
      if (client != null && !client.isConnected()) {
        try {
          log.info("开始第" + (i + 1) + "次连接");
          client.reconnect(); // 重连
        } catch (MqttException e) {
          log.error("重新连接消息机失败:" + e);
        }
      }
      try {
        Thread.sleep(1000);// 休眠1s
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      if (client != null && client.isConnected()) {
        long end = System.currentTimeMillis();
        log.info("连接成功:共使用了:" + (end - start) + "ms");
        return;
      }
    }
  }

  /**
   * 订阅主题
   * 
   * @param topic
   * @param qos
   * @return
   */
  public boolean subscribeTopic(String topic, int qos) {
    boolean result = false;
    if (client != null && client.isConnected()) {
      try {
        client.subscribe(topic, qos);
        result = true;
      } catch (MqttException e) {
        e.printStackTrace();
      }
    }
    return result;
  }
}